import java.io.File;
import java.io.IOException;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.Map;


/**
 * Class representing a relational algebra operation to order a table based on its attributes.
 * @author Jonathan
 *
 */
public class RAOrder implements RAExpression, Cloneable {
	
	private Map<String, TableData> catalog;
	private RAExpression inputRA;
	private ArrayList<Expression> select;
	private Map<String, String> from;
	private ArrayList<SortExpression> orderBy;

	public RAOrder(Map<String, TableData> catalog, ArrayList<Expression> select, Map<String, String> from,
			ArrayList<SortExpression> orderBy, RAExpression inputRA)
	{
		this.catalog = catalog;
		this.select = select;
		this.from = from;
		this.orderBy = orderBy;
		this.inputRA = inputRA;
	}
	
	protected RAOrder clone()
	{
		try
		{
			return (RAOrder) super.clone();
		}
		catch(Exception e)
		{
			throw new RuntimeException(e);
		}
	}

	@Override
	public RAOutput execute() throws IOException
	{
		// Execute the input RA
		RAOutput raOutput = inputRA.execute();
		
		// Prepare the input information
		ArrayList<Attribute> inAtts = raOutput.outAtts;
		String inputFile = raOutput.outFile;
		final String outputFile = inputFile.replaceAll("\\.[sjg]out", "").replaceAll("\\.tbl", "") + ".oout";
		
		// Prepare output Attributes
		int selectSize = select.size();
		ArrayList<Attribute> outAtts = new ArrayList<Attribute>(selectSize);
		Map<String, String> exprsToCompute = new HashMap<String, String>();
		for(int i = 0; i < selectSize; i++)
		{
			Expression e = select.get(i);
			String attName = "att" + i;
			String returnType = e.getReturnType(from, catalog);
			String executionString = e.toExecutionString();
			
			outAtts.add(new Attribute(returnType, attName));
			exprsToCompute.put(attName, executionString);
		}
		
		// Prepare sorting information
		int sortKeyCount = orderBy.size();
		ArrayList<SortKeyExpression> sortingKeys = new ArrayList<SortKeyExpression>(sortKeyCount);
		for(int i = 0; i < sortKeyCount; i++)
		{
			SortExpression se = orderBy.get(i);
			
			Expression e = se.getExpression();
			boolean direction = se.getDirections();
			
			String dataType = e.getReturnType(from, catalog);
			String executionString = e.toExecutionString();
			
			sortingKeys.add(new SortKeyExpression(dataType, executionString, direction));
		}
		
		try
		{
			new Order(inAtts, outAtts, exprsToCompute, sortingKeys, inputFile, outputFile, DBConfig.CPP_COMPILER, DBConfig.CPP_TEMP_DIR);
		}
		catch (Exception e)
		{
			throw new RuntimeException(e);
		}
		
		// This has been moved to Order
//		if(inputRA != null && !inputFile.endsWith(".tbl"))
//		{
//			File f = new File(inputFile);
//			f.delete();
//		}
		
		return new RAOutput(outputFile, outAtts);
	}

	@Override
	public void pushdownReferencedColumnsList(ArrayList<Expression> selectList)
	{
		// Ordering will always be the last thing that happens, so it should
		// just have the order list for the expressions and order them without
		// any complicated selection/project.
		inputRA.pushdownReferencedColumnsList(selectList);
	}

	@Override
	public <T> T accept(RAExpressionVisitor<T> v, Object... args)
	{
		return v.orderCase(this, args);
	}

	@Override
	public RAStatistics estimateCost(Map<String, TableData> catalog,
			Map<String, TableData> tempCatalog)
	{
		RAStatistics stats = inputRA.estimateCost(catalog, tempCatalog);
		long inputCard = stats.getCardinality();
		long inputCost = stats.getCost();

		// Ordering does one more round of io and sorting and increases the cost by inputCard.
		// Sorting should run in O(n log n) so use n log n as the cost.
		return new RAStatistics(inputCard, inputCost + (inputCard * (long) Math.log(inputCard)));
	}

	@Override
	public void getTables(Map<String, String> tables)
	{
		tables.putAll(from);
//		inputRA.getTables(tables);
	}

	@Override
	public ArrayList<RAExpression> getJoinTransformations()
	{
		ArrayList<RAExpression> joinTransformations = inputRA.getJoinTransformations();
		if(joinTransformations != null)
		{
			int transformCount = joinTransformations.size();
			for(int i = 0; i < transformCount; i++)
			{
				RAExpression ra = joinTransformations.get(i);
				RAOrder myClone = clone();
				myClone.inputRA = ra;
				joinTransformations.set(i, myClone);
			}
			return inputRA.getJoinTransformations();
		}
		return null;
	}

	@Override
	public boolean referencedByExpression(Expression e)
	{
		return inputRA.referencedByExpression(e);
	}

	@Override
	public void addWhereDisjunction(Expression e)
	{
		throw new RuntimeException("Cannot currently add a where disjunction to an orderBy");
	}

	@Override
	public String joinPlanToString()
	{
		return "order(" + inputRA.joinPlanToString() + ")";
	}

	@Override
	public void applySelectList(ArrayList<Expression> select)
	{
		
	}

	@Override
	public void pushdownJoinPredicates(ArrayList<Expression> joinWhereClauses)
	{
		inputRA.pushdownJoinPredicates(joinWhereClauses);
	}

}
